package org.example.testKafka.service;

import cn.hutool.json.JSONUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

@Slf4j
@Component
@RequiredArgsConstructor
public class Consumer2 {

    @KafkaListener(topics = "ylty-schema-nlp", properties = "${spring.kafka.consumer.group-id2}")
    public void insertTopicAccount(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
        if (!records.isEmpty()) {

            for (ConsumerRecord<String, String> record : records) {
                String jsonStr = null;
                try {
                    jsonStr = record.value();
                } catch (Exception e) {
                    log.error("插入账号报错：{}", jsonStr, e);
                }
            }
            ack.acknowledge();
        }
    }

}
